package com.intct.flink.study;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.kafka.clients.producer.ProducerConfig;

/**
 * @author gufg
 * @since 2025-10-10 15:57
 */
public class CdcDSApp {
    public static void main(String[] args) throws Exception {
        // 获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并度
//        env.setParallelism(1);
        // 开启检查
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 配置源--mysql cdc
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("cdh-node")
                .port(13306)
                .databaseList("m1") // 设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
                .tableList("m1.test05") // 设置捕获的表
                .username("root")
                .password("Test_090110")
                .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
                /*启动模式
                    配置选项scan.startup.mode指定 MySQL CDC 使用者的启动模式。有效枚举包括：
                    initial （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。
                    earliest-offset：跳过快照阶段，从可读取的最早 binlog 位点开始读取
                    latest-offset：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
                    specific-offset：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。
                    timestamp：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。
                */
//                .startupOptions(StartupOptions.latest())
                .chunkKeyColumn(new ObjectPath("m1", "test05"), "id")
                .build();

        // 接收数据
        DataStreamSource<String> sourceDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "cdc_ds_app_source");


        // 输出
        // sourceDS.print();

        //Kafka配置，要求：使用精准一次语义
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                .setBootstrapServers("cdh-node:9092")
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic("datasteam_api_topic")
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                /*
                容错 #
                KafkaSink 总共支持三种不同的语义保证（DeliveryGuarantee）。对于 DeliveryGuarantee.AT_LEAST_ONCE 和
                DeliveryGuarantee.EXACTLY_ONCE，Flink checkpoint 必须启用。默认情况下 KafkaSink 使用 DeliveryGuarantee.NONE。 以下是对不同语义保证的解释：

                DeliveryGuarantee.NONE 不提供任何保证：消息有可能会因 Kafka broker 的原因发生丢失或因 Flink 的故障发生重复。
                DeliveryGuarantee.AT_LEAST_ONCE: sink 在 checkpoint 时会等待 Kafka 缓冲区中的数据全部被 Kafka producer 确认。
                                                消息不会因 Kafka broker 端发生的事件而丢失，但可能会在 Flink 重启时重复，因为 Flink 会重新处理旧数据。
                DeliveryGuarantee.EXACTLY_ONCE: 该模式下，Kafka sink 会将所有数据通过在 checkpoint 时提交的事务写入。
                                                   因此，如果 consumer 只读取已提交的数据（参见 Kafka consumer 配置 isolation.level），
                                                   在 Flink 发生重启时不会发生数据重复。然而这会使数据在 checkpoint 完成时才会可见，因此请按需调整 checkpoint 的间隔。
                                                   请确认事务 ID 的前缀（transactionIdPrefix）对不同的应用是唯一的，以保证不同作业的事务 不会互相影响！
                                                   此外，强烈建议将 Kafka 的事务超时时间调整至远大于 checkpoint 最大间隔 + 最大重启时间，
                                                   否则 Kafka 对未提交事务的过期处理会导致数据丢失。
               */
                /*
                    设置精准一语义时，需要同时
                        1、开始检查点(checkpoint)
                        2、事务超时间
                        3、事务ID的前缀
                 */
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                // 设置精准一次语义时，需要设置事务超时间
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "90000")
                .setTransactionalIdPrefix("ds_id_")
                .build();

        // 输出数据到Kafka
        sourceDS.sinkTo(kafkaSink);

        // 启动作业
        env.execute("cdc_ds_app_source");
    }







}
